package com.atguigu.chapter11;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName: Flink06_Table_SQL_BaseUse
 * @Description:
 * @Author: kele
 * @Date: 2021/4/12 20:43
 *
 * 基本操作
 *
 *
 **/
public class Flink06_Table_SQL_BaseUse {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        DataStreamSource<WaterSensor> waterSensorStream =
                env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60));

        Table table = tenv.fromDataStream(waterSensorStream);  // 动态表

        tenv.createTemporaryView("tt",table);  //给读取的表创建一个临时的表名

        //tenv.executeSql() // 执行增删改(DML) 创建表(DDL) 的这些sql
        // 执行查询的sql
        // 查询一个未注册的表
        tenv
             //   .sqlQuery("select * from " + table + " where id='sensor_1'")
                .sqlQuery("select * from tt where id='sensor_1'")
                .execute()
                .print();


    }

}
